
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.io.IOException;
import java.util.Arrays;
import java.util.Iterator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;

public class HbaseTabManger {

	/**
	 * 配置ss
	 Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "emr-worker-2,emr-worker-1,emr-header-1");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.master", "127.0.0.1:60000");
        Connection conn = ConnectionFactory.createConnection(configuration);
        Admin admin = conn.getAdmin();
	 */
	static Configuration config = null;
	private Connection connection = null;
	private Table table = null;
        
	public void init() throws Exception {
		config = HBaseConfiguration.create();// 配置
		config.set("hbase.zookeeper.quorum", "emr-worker-2,emr-worker-1,emr-header-1");// zookeeper地址
		config.set("hbase.zookeeper.property.clientPort", "2181");// zookeeper端口
		//config.set("hbase.master", "127.0.0.1:60000");
		connection = ConnectionFactory.createConnection(config);
		Admin admin = connection.getAdmin();
		//验证表是否存在，是则删除
		TableName tableName = TableName.valueOf("zimo:student");
                if (admin.tableExists(tableName)) {
                    //deleteTable(tableName);
                }else { 
                   createTable();
                }
	        table = connection.getTable(tableName);
	}

	/**
	 * 创建数据库表并增加列族info和score
	 *
	 * @throws Exception
	 */
	public void createTable() throws Exception {
		// 创建表管理类
		Admin admin = connection.getAdmin(); // hbase表管理
		// 创建表描述类
		TableName tableName = TableName.valueOf("zimo:student"); // 表名称
		HTableDescriptor desc = new HTableDescriptor(tableName);
		
		HColumnDescriptor info = new HColumnDescriptor("info"); // 列族
		// 将列族添加到表中
		desc.addFamily(info);
		HColumnDescriptor score = new HColumnDescriptor("score"); // 列族
		// 将列族添加到表中
		desc.addFamily(score);
		// 创建表
		admin.createTable(desc); // 创建表
		System.out.println("创建表成功！");
	}

	
	public void insertData(String key,String frist_col,String second_col,String colValue) throws Exception {
	     // table.setAutoFlushTo(false);
	      Put put = new Put(Bytes.toBytes(key));//创建rowkey
	      put.addColumn(Bytes.toBytes(frist_col), Bytes.toBytes(second_col), Bytes.toBytes(colValue));
	      //插入数据
	      table.put(put);
	      //提交
	      //table.flushCommits();
	      System.out.println(key+"数据插入成功！");
	}
	
	
	

	
	public void selectData(String frist_col,String second_col) throws Exception {

		// 创建全表扫描的scan
		Scan scan = new Scan();
		System.out.println("查询的数据");
		// 打印结果集
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
                      List<Cell> cells = result.listCells();
                      //迭代所有的cell
                      for (Cell c : cells) {
                            //获取行键
                         byte[] rowArray = c.getRowArray();
                         System.out.print(" info = " + rowArray + "\n");
                      }
                /*
			if (result.getValue(Bytes.toBytes(frist_col), Bytes.toBytes(second_col)) == null) {
				for (KeyValue kv : result.raw()) {
				     System.out.print(new String(kv.getRow()) + " ");
				     System.out.print(new String(kv.getFamily()) + ":");
				     System.out.print(new String(kv.getQualifier()) + " = ");
				     System.out.print(new String(kv.getValue()));
				     System.out.print(" info = " + kv.getTimestamp() + "\n");
				}
		}*/

		}
	}

	

	public void close() throws Exception {
		table.close();
		connection.close();
	}
	public static void main(String[] args) throws Exception{
	  HbaseTabManger hm = new HbaseTabManger();
	  hm.init();
	  hm.insertData("zimo","score","understanding","89");
	  hm.selectData("score","understanding");
	}

}
